import PageView.getClass
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * @author: yangShen
  * @Description: UV统计，独立用户访问                          -----网站独立访客数UV的统计
  * @Date: 2020/4/29 22:18 
  */

case class UvCount(windowEnd: Long, uvCount: Long)

object UniqueVisitor {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //用相对路径定义数据源
    val resource = getClass.getResource("/UserBehavior.csv")
    val dataStream = environment.readTextFile(resource.getPath)
      .map(data => {
        val dataArray = data.split(",")
        UserBehavior(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim.toInt, dataArray(3).trim, dataArray(4).trim.toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)
      .filter(_.behavior == "pv")   //只统计pv操作
      //在窗口内去重
      .timeWindowAll(Time.hours(1))
      .apply(new UvCountByWindow())

    dataStream.print()

    environment.execute("UV count")

  }
}
//AllWindowFunction比WindowFunction少了一个key的类型, 因为AllWindowFunction之前没有做keyBy操作
class UvCountByWindow() extends AllWindowFunction[UserBehavior, UvCount, TimeWindow]{
  override def apply(window: TimeWindow, input: Iterable[UserBehavior], out: Collector[UvCount]): Unit = {
    // 定义一个scala set, 用于保存所有数据的userId并去重(把放到set中，利用set特性去重),如果数据量非常的大 可以把去重放到redis中做
    var idSet = Set[Long]()
    //把当前窗口所有数据的ID收集到set中，最后输出set的大小
   for(userBehavior <- input ){
     idSet += userBehavior.userId
   }

   out.collect(UvCount(window.getEnd, idSet.size))
  }
}
